package com.uuuo.rxjava1;

import org.junit.Test;

import java.util.concurrent.TimeUnit;

import rx.Observable;
import rx.Subscriber;
import rx.functions.Action1;
import rx.functions.Func0;

/**
 * Created by JiaTao on 2017/4/1.
 */

public class A_Observable_Create {
    public void showMsg(String str) {
        System.out.println(str);
    }

    @Test
    public void Create() {
        Observable.create(new Observable.OnSubscribe<String>() {
            @Override
            public void call(Subscriber<? super String> subscriber) {
                subscriber.onNext("jia");
                subscriber.onNext("tao");
                subscriber.onNext("!");
                subscriber.onCompleted();
            }
        }).subscribe(new Subscriber<String>() {
            @Override
            public void onCompleted() {
                showMsg("onCompleted");
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(String s) {
                showMsg(s);
            }
        });
    }


    int number = 0;

    @Test
    public void Defer() {
        number = 1;
        Observable justObservable = Observable.just(number);

        Observable createObservable = Observable.create(new Observable.OnSubscribe<Integer>() {
            @Override
            public void call(Subscriber<? super Integer> subscriber) {
                subscriber.onNext(number);
            }
        });

        number = 2;
        Observable deferObservable = Observable.defer(new Func0<Observable<Integer>>() {
            @Override
            public Observable<Integer> call() {
                return Observable.just(number);
            }
        });

        number = 3;

        justObservable.subscribe(new Action1() {
            @Override
            public void call(Object o) {
                System.out.println("just result: " + o.toString());
            }
        });

        createObservable.subscribe(new Action1() {
            @Override
            public void call(Object o) {
                System.out.println("create result: " + o.toString());
            }
        });

        deferObservable.subscribe(new Action1() {
            @Override
            public void call(Object o) {
                System.out.println("defer1 result: " + o.toString());
            }
        });

        number = 4;

        deferObservable.subscribe(new Action1() {
            @Override
            public void call(Object o) {
                System.out.println("defer2 result: " + o.toString());
            }
        });
    }

    @Test
    public void From() {
        Integer[] ints = {1, 3, 5, 7, 9};
        Observable.from(ints).subscribe(new Action1<Integer>() {
            @Override
            public void call(Integer integer) {
                showMsg(integer + "");
            }
        });
    }

    @Test
    //创建一个按固定时间间隔发射整数序列的Observable。
    public void Interval() {
//        Observable.interval(5, TimeUnit.SECONDS)
//                .take(5)
//                .subscribe(new Subscriber<Long>() {
//                    @Override
//                    public void onCompleted() {
//                        showMsg("Interval onCompleted");
//                    }
//
//                    @Override
//                    public void onError(Throwable e) {
//                        e.printStackTrace();
//                    }
//
//                    @Override
//                    public void onNext(Long aLong) {
//                        showMsg(aLong + "");
//                    }
//                });


        Observable observable = Observable.interval(1000, TimeUnit.MILLISECONDS);

        Subscriber subscriber = new Subscriber<Long>() {

            @Override

            public void onCompleted() {
                System.out.println("onCompleted.");
            }


            @Override

            public void onError(Throwable e) {
                System.out.println("onError: " + e.getMessage());
            }


            @Override

            public void onNext(Long aLong) {
                System.out.println("onNext: " + aLong);
            }
        };

        observable.subscribe(subscriber);
    }

    @Test
    public void Repeat() {
        Observable.range(2, 3)
                .repeat(2)
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onCompleted() {
                        System.out.println("onCompleted.");
                    }
                    @Override
                    public void onError(Throwable e) {
                        System.out.println("onError: " + e.getMessage());
                    }
                    @Override
                    public void onNext(Integer integer) {
                        System.out.println("onNext: " + integer);
                    }
                });
    }

}
